package com.dbms.common;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
/**
 * `Tuple` 类代表表中的一行数据（也称为记录）。
 * 它内部以紧凑的字节数组 (`data`) 形式存储数据，并通过 `Schema` 来解释这些字节的含义。
 * 这个类负责数据的序列化（从Java对象列表到字节数组）和反序列化（从字节数组到Java对象）。
 */


public class Tuple implements Serializable {
    private final byte[] data;
    // 描述这个元组结构的Schema。没有它，`data` 字节数组就毫无意义。
    private final Schema schema;

    // 从字节数组和Schema构造元组（反序列化）
    /**
     * 构造函数（用于反序列化）：从已有的字节数组和Schema构造一个元组。
     * 当从磁盘页面中读取数据时使用此构造函数。
     * @param data   包含元组数据的字节数组。
     * @param schema 描述该元组结构的Schema。
     */
    public Tuple(byte[] data, Schema schema) {
        this.data = data;
        this.schema = schema;
    }

    public Schema getSchema() {
        return schema;
    }

    // 从字段值列表和Schema构造元组（序列化）
    /**
     * 构造函数（用于序列化）：从一个Java对象列表和Schema构造一个元组。
     * 当需要将新数据（例如来自INSERT语句）存入数据库时使用此构造函数。
     * @param values 一个包含各个字段值的列表，其顺序和类型必须与Schema匹配。
     * @param schema 描述该元组结构的Schema。
     */
    public Tuple(List<Object> values, Schema schema) {
        this.schema = schema;
        // 序列化过程
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos);
        try {
            for (int i = 0; i < schema.getColumnCount(); i++) {
                Column col = schema.getColumn(i);
                if (col.getColumnType() == Type.INTEGER) {
                    dos.writeInt((Integer) values.get(i));
                } else if (col.getColumnType() == Type.VARCHAR) {
                    String str = (String) values.get(i);
                    byte[] strBytes = str.getBytes();
                    dos.writeInt(strBytes.length); // 先写入字符串长度
                    dos.write(strBytes);           // 再写入字符串内容
                }
            }
            dos.flush();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.data = baos.toByteArray();
    }

    public byte[] getData() {
        return data;
    }

    public Object getValue(int fieldIndex) {
        Column col = schema.getColumn(fieldIndex);
        ByteBuffer buffer = ByteBuffer.wrap(data);

        // 需要遍历找到正确的偏移量
        int offset = 0;
        for (int i = 0; i < fieldIndex; i++) {
            Column prevCol = schema.getColumn(i);
            if (prevCol.getColumnType() == Type.INTEGER) {
                offset += 4;
            } else if (prevCol.getColumnType() == Type.VARCHAR) {
                int len = buffer.getInt(offset);
                offset += 4 + len;
            }
        }

        if (col.getColumnType() == Type.INTEGER) {
            return buffer.getInt(offset);
        } else if (col.getColumnType() == Type.VARCHAR) {
            int len = buffer.getInt(offset);
            byte[] strBytes = new byte[len];
            // 从buffer的 offset + 4 位置开始读
            System.arraycopy(data, offset + 4, strBytes, 0, len);
            return new String(strBytes);
        }
        return null;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder("Tuple{");
        for (int i = 0; i < schema.getColumnCount(); i++) {
            sb.append(getValue(i)).append(i == schema.getColumnCount() - 1 ? "" : ", ");
        }
        sb.append("}");
        return sb.toString();
    }
}

